package com.boot.message.rabbit;

import cn.hutool.core.util.IdUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.nio.charset.StandardCharsets;
import java.util.Objects;

/**
 * @author 霜寒 <1621856595@qq.com>
 * @description RabbitMQ配置
 * @date 2020/4/30 0:58
 **/
@Slf4j
@Configuration
@EnableRabbit
public class RabbitMQConfig implements InitializingBean {

    @Bean("bootExchange")
    public Exchange bootExchange() {
        return ExchangeBuilder.topicExchange("bootExchange").durable(true).build();
    }

    @Bean("userQueue")
    public Queue userQueue() {
        return QueueBuilder.durable("userQueue").build();
    }

    @Bean("groupQueue")
    public Queue groupQueue() {
        return QueueBuilder.durable("groupQueue").build();
    }

    @Bean
    public Binding userBinding(@Qualifier("userQueue") Queue queue,
                               @Qualifier("bootExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("user").noargs();
    }

    @Bean
    public Binding groupBinding(@Qualifier("groupQueue") Queue queue,
                                @Qualifier("bootExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("group").noargs();
    }

    Jackson2JsonMessageConverter jackson2JsonMessageConverter(ObjectMapper mapper) {
        return new Jackson2JsonMessageConverter(mapper);
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        //        rabbitAdmin.setIgnoreDeclarationExceptions(true);
        return rabbitAdmin;
    }

    @Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        //同时监听多个队列
        container.setQueues();
        //设置当前的消费者数量
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(5);
        //设置是否重回队列
        container.setDefaultRequeueRejected(false);
        //设置自动签收
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        //设置监听外露
        container.setExposeListenerChannel(true);
        //设置消费端标签策略
        container.setConsumerTagStrategy(queue -> queue + "_" + IdUtil.fastSimpleUUID());
        //设置消息监听
        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            log.info("消费者：{}", msg);
        });
        return container;
    }

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    ObjectMapper objectMapper;

    @Override
    public void afterPropertiesSet() {
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter(objectMapper));
        rabbitTemplate.setEncoding("UTF-8");
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnsCallback(returnedMessage -> log.debug("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {}  路由键: {}",
                returnedMessage.getMessage().getMessageProperties().getCorrelationId(),
                returnedMessage.getReplyCode(),
                returnedMessage.getReplyText(),
                returnedMessage.getExchange(),
                returnedMessage.getRoutingKey()));
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.debug("消息发送成功,id: {}", Objects.requireNonNull(correlationData).getId());
            } else {
                log.debug("消息发送失败,原因: {}", cause);
            }
        });
    }
}
